package com.atguigu.flink.day05;

import com.atguigu.flink.beans.WaterSensor;
import com.atguigu.flink.func.WaterSensorMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author Felix
 * @date 2023/12/5
 * 该案例演示了窗口处理函数-reduce
 */
public class Flink01_window_reduce {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> wsDS = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        //如果没有对流中数据进行keyby，可以调用.countWindowAll() 或者.windowAll()进行开窗
        //这时是对整条流数据进行开窗聚合计算，开窗之后，并行度变为1

        /*KeyedStream<WaterSensor, String> keyedDS = wsDS.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor ws) throws Exception {
                return ws.id;
            }
        });
        wsDS.keyBy(ws->ws.id)
        */
        KeyedStream<WaterSensor, String> keyedDS = wsDS.keyBy(WaterSensor::getId);

        //如果对流进行keyby之后，可以调用.window() 或者是.countWindow()进行开窗
        //这时是对keyby之后的每组数据进行单独的开窗，组和组之间相互并不影响
        //window算子参数：WindowAssigner---窗口分配器（指点开一个什么样的窗口--窗口类型）
        //TumblingProcessingTimeWindows.of(Time.seconds(10)) 开一个滚动处理时间窗口，窗口大小是10s
        WindowedStream<WaterSensor, String, TimeWindow> windowDS = keyedDS
            .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        //使用reduce算子对窗口中的数据进行增量聚合计算
        //reduce: 窗口中的元素类型和聚合之后向下游传递的元素类型必须一致
        //当窗口中只有一条数据的时候，reduce方法不会被调用，直接将这条数据向下游传递
        windowDS.reduce(
            new ReduceFunction<WaterSensor>() {
                @Override
                public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                    System.out.println("中间结果：" + value1);
                    System.out.println("新进数据：" + value2);
                    return new WaterSensor(value1.id,System.currentTimeMillis(),value1.vc + value2.vc);
                }
            }
        ).print("~~~");

        env.execute();
    }
}
